package com.jd.jobkeeper.server.transport;

import com.google.protobuf.MessageLite;
import com.jd.jobkeeper.protocals.SetupRequest;
import com.jd.jobkeeper.server.configure.ServerConfig;
import com.jd.jobkeeper.transport.HandlerRegistry;
import com.jd.jobkeeper.utils.InvokerHandler;
import com.jd.jobkeeper.utils.ProtobufDecoder;
import com.jd.jobkeeper.utils.ProtobufEncoder;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.SocketAcceptor;
import io.rsocket.core.RSocketServer;
import io.rsocket.transport.ServerTransport;
import io.rsocket.transport.netty.server.CloseableChannel;
import io.rsocket.transport.netty.server.TcpServerTransport;
import io.rsocket.util.DefaultPayload;
import io.rsocket.util.RSocketProxy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

@Component
public class TcpServer {
    static final Logger logger = LoggerFactory.getLogger(TcpServer.class);
    private CloseableChannel channel;
    private ProtobufDecoder decoder = new ProtobufDecoder();
    private ProtobufEncoder encoder = new ProtobufEncoder();

    private HandlerRegistry registry = new HandlerRegistry();

    private final ServerConfig serverConfig;

    public TcpServer(ServerConfig serverConfig) {
        this.serverConfig = serverConfig;
    }

    public void start() {
        this.channel = RSocketServer.create(createSocketAcceptor())
                .bind(createTcpTransport())
                //.delaySubscription(Duration.ofSeconds(5))
                .doOnNext(cc -> logger.info("Server started on the address : {}", cc.address()))
                .block();
    }

    public void registerHandler(Object proxy, Class bizTypeInterface) {
        registry.register(proxy, bizTypeInterface);
    }

    private ServerTransport<CloseableChannel> createTcpTransport() {
        reactor.netty.tcp.TcpServer tcpServer = reactor.netty.tcp.TcpServer.create()
                .port(this.serverConfig.getPort());
        return TcpServerTransport.create(tcpServer);
    }

    private SocketAcceptor createSocketAcceptor() {
        return (setupPayload, sendingSocket) -> {
            SetupRequest setupRequest = null;
            if (setupPayload.hasMetadata()) {
                setupPayload.retain();
                setupRequest = (SetupRequest) decoder.decode(setupPayload, SetupRequest.class);
            }
            return Mono.just(new InternalRSocket(setupRequest, sendingSocket));
        };
    }

    class InternalRSocket extends RSocketProxy {
        final com.jd.jobkeeper.server.transport.ServerTransport transport;

        public InternalRSocket(SetupRequest setupRequest, RSocket sendingSocket) {
            super(sendingSocket);
            this.transport = new com.jd.jobkeeper.server.transport.ServerTransport(setupRequest,sendingSocket);
        }

        @Override
        public Mono<Payload> requestResponse(Payload payload) {
            try {
                String meta = payload.getMetadataUtf8();
                InvokerHandler invokerHandler = registry.lookup(meta);
                Class<?> parameterTypes[] = invokerHandler.getParameterType();
                Object arg = decoder.decode(payload, parameterTypes[0]);
                return invokerHandler.invoke(arg,transport)
                        .map(data -> DefaultPayload.create(encoder.encode((MessageLite) data)));
            } finally {
                payload.release();
            }
        }

        /*@Override
        public Mono<Void> fireAndForget(Payload payload) {
            try {
                String meta = payload.getMetadataUtf8();
                InvocationHandler invocationHandler = registry.lookup(meta);
                Class<?> parameterTypes[] = invocationHandler.getParameterType();
                Object arg = decoder.decode(payload, parameterTypes[0]);
                return invocationHandler.invoke(arg,transport)
                        .map(data -> DefaultPayload.create(encoder.encode((MessageLite) data)));
            } finally {
                payload.release();
            }
        }*/
    }

}
